package com.zlcx.tz.live.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.zlcx.tz.live.client.UserClient;
import com.zlcx.tz.live.client.vo.LiveGatherVO;
import com.zlcx.tz.live.client.vo.UserFriendVO;
import com.zlcx.tz.live.constant.PublisherStatus;
import com.zlcx.tz.live.constant.RedisKeyConst;
import com.zlcx.tz.live.constant.YN;
import com.zlcx.tz.live.domain.LiveFree;
import com.zlcx.tz.live.domain.LiveRoomInfo;
import com.zlcx.tz.live.domain.LiveRoomJoinRecord;
import com.zlcx.tz.live.domain.LiveRoomPublisherRecord;
import com.zlcx.tz.live.domain.backup.LiveRoomPublisherRecordBackUpDO;
import com.zlcx.tz.live.enums.RoomMicStatus;
import com.zlcx.tz.live.enums.RoomState;
import com.zlcx.tz.live.enums.RoomUserRole;
import com.zlcx.tz.live.exception.BusinessException;
import com.zlcx.tz.live.exception.CheckParamException;
import com.zlcx.tz.live.repository.LiveRoomPublisherRecordRepository;
import com.zlcx.tz.live.repository.backup.LiveRoomPublisherRecordBackupRepository;
import com.zlcx.tz.live.vo.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.data.domain.Example;
import org.springframework.data.domain.Sort;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import javax.annotation.Resource;
import javax.persistence.criteria.CriteriaBuilder;
import javax.persistence.criteria.CriteriaQuery;
import javax.persistence.criteria.Predicate;
import javax.persistence.criteria.Root;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static com.zlcx.tz.live.constant.PublisherStatus.PUBLISHING;

/**
 * <p></p>
 * <PRE>
 * <BR>    修改记录
 * <BR>-----------------------------------------------
 * <BR>    修改日期         修改人          修改内容
 * </PRE>
 *
 * @author zl
 * @version 1.0
 * @date Created in 2019/4/18 10:42
 * @copyright: Copyright (c) founders
 */
@Service
@Slf4j
public class LiveRoomPublisherService {

    @Autowired
    LiveRoomPublisherRecordRepository publisherRecordRepository;

    @Autowired
    MessageService messageService;

    @Autowired
    UserService userService;

    @Autowired
    GroupService groupService;

    @Autowired
    RedisTemplate redisTemplate;

    @Resource(name = "taskExecutor")
    AsyncTaskExecutor asyncTaskExecutor;

    @Autowired
    LiveRoomService roomService;

    @Autowired
    private UserClient userClient;

    @Autowired
    LiveRoomJoinRecordService roomJoinRecordService;

    @Autowired
    private LiveRoomPublisherRecordBackupRepository liveRoomPublisherRecordBackupRepository;

    @Value("${sunlands.liveTimeOut}")
    private Integer liveTimeOut;

    private Logger traceLoger = LoggerFactory.getLogger("traceloger");

    /**
     * 房主创建房间之后直接上麦
     *
     * @param roomId
     * @param userId
     */
    public RoomPublisherVO hostApplyPublish(Long roomId, Long userId, String studyContent) {

        LiveRoomPublisherRecord liveRoomPublisherRecord = new LiveRoomPublisherRecord();
        liveRoomPublisherRecord.setRoomId(roomId);
        liveRoomPublisherRecord.setUserId(userId);
        Date now = new Date();
        liveRoomPublisherRecord.setApplyTime(now);
        liveRoomPublisherRecord.setAgreeTime(now);
        liveRoomPublisherRecord.setPublisherState(PUBLISHING.getCode());
        liveRoomPublisherRecord.setCreateTime(now);
        liveRoomPublisherRecord.setUpdateTime(now);
        liveRoomPublisherRecord.setStudyContent(studyContent);

        liveRoomPublisherRecord = publisherRecordRepository.save(liveRoomPublisherRecord);
        traceLoger.info("hostApplyPublish 房间【{}】房主【{}】上麦【{}】", roomId, userId, liveRoomPublisherRecord.getId());

        LiveRoomPublisherRecord finalLiveRoomPublisherRecord = liveRoomPublisherRecord;
        asyncTaskExecutor.execute(() -> {
            liveTask(userId, roomId);
            String publisherCountKey = RedisKeyConst.getRoomPublisherCountKey(roomId.toString());
            Long count = redisTemplate.opsForValue().increment(publisherCountKey, 1L);
            if (count != 1) {
                redisTemplate.opsForValue().set(publisherCountKey, 1L);
            }
            redisTemplate.expire(publisherCountKey, 2, TimeUnit.DAYS);
            String currentPublishingRecordKey = RedisKeyConst.getCurrentPublishingRecordKey(userId, roomId);
            redisTemplate.opsForValue().set(currentPublishingRecordKey, finalLiveRoomPublisherRecord, 2, TimeUnit.DAYS);
            startLearnTask(userId, roomId, finalLiveRoomPublisherRecord.getId(), studyContent);
        });

        RoomPublisherVO roomPublisherVO = new RoomPublisherVO();
        BeanUtils.copyProperties(liveRoomPublisherRecord, roomPublisherVO);
        return roomPublisherVO;
    }

    /**
     * 上麦申请
     *
     * @param roomId
     * @param userId
     */
    public RoomPublisherVO applyPublish(Long roomId, Long userId, String studyContent) {

        UserVO userVO = getUserVO(userId);
        if (userVO == null) {
            traceLoger.info("applyPublish 房间【{}】用户【{}】申请上麦，用户不存在", roomId, userId);
            log.error("[applyPublish]用户不存在，roomId={},userId={}", roomId, userId);
            throw new CheckParamException("用户不存在!");
        }

        LiveRoomInfo roomInfo = roomService.getRoomInfoById(roomId);
        if (roomInfo == null) {
            traceLoger.info("applyPublish 房间【{}】用户【{}】申请上麦，房间不存在", roomId, userId);
            log.error("[applyPublish]房间不存在，roomId={},userId={}", roomId, userId);
            throw new CheckParamException("房间不存在或关闭!");
        }

        if (roomJoinRecordService.getUserJoinByRoomIdAndUserId(roomId, userId) == null) {
            traceLoger.info("applyPublish 房间【{}】用户【{}】申请上麦，加入此房间记录不存在", roomId, userId);
            throw new CheckParamException("不在此房间中，请重新进入!");
        }

        LiveRoomPublisherRecord oldRecord = getCurrentPublishingRecord(roomId, userId);
        if (oldRecord != null && PublisherStatus.isValidStatus(oldRecord.getPublisherState())) {
            traceLoger.info("applyPublish 房间【{}】用户【{}】申请上麦，上麦申请已经存在或者在上麦中，上麦id【{}】", roomId, userId, oldRecord.getId());
            log.error("[applyPublish]上麦申请已经存在，roomId={},userId={}", roomId, userId);
            throw new BusinessException("非法操作!");
        }

        long publishingCount = getPublishingCount(roomId);
        if (publishingCount >= roomInfo.getRoomPublisherMax()) {
            traceLoger.info("applyPublish 房间【{}】用户【{}】申请上麦，房间主播已经满员", roomId, userId);
            throw new CheckParamException("当前房间主播已经满员!");
        }

        LiveRoomPublisherRecord liveRoomPublisherRecord = new LiveRoomPublisherRecord();
        liveRoomPublisherRecord.setRoomId(roomId);
        liveRoomPublisherRecord.setUserId(userId);
        Date now = new Date();
        liveRoomPublisherRecord.setApplyTime(now);
        liveRoomPublisherRecord.setPublisherState(PublisherStatus.APPLYING.getCode());
        liveRoomPublisherRecord.setCreateTime(now);
        liveRoomPublisherRecord.setUpdateTime(now);
        liveRoomPublisherRecord.setStudyContent(studyContent);

        liveRoomPublisherRecord = publisherRecordRepository.save(liveRoomPublisherRecord);
        traceLoger.info("applyPublish 房间【{}】用户【{}】申请上麦【{}】", roomId, userId, liveRoomPublisherRecord.getId());

        messageService.applyPublishMsg(roomId, liveRoomPublisherRecord.getId(), userVO, now);

        RoomPublisherVO roomPublisherVO = new RoomPublisherVO();
        BeanUtils.copyProperties(liveRoomPublisherRecord, roomPublisherVO);
        return roomPublisherVO;
    }


    /**
     * 当前上麦记录
     *
     * @param roomId
     * @param userId
     * @return
     */
    public LiveRoomPublisherRecord getCurrentPublishingRecord(Long roomId, Long userId) {

        String currentPublishingRecordKey = RedisKeyConst.getCurrentPublishingRecordKey(userId, roomId);
        LiveRoomPublisherRecord liveRoomPublisherRecord = (LiveRoomPublisherRecord) redisTemplate.opsForValue().get(currentPublishingRecordKey);
        if (liveRoomPublisherRecord != null) {
            return liveRoomPublisherRecord;
        }

        liveRoomPublisherRecord = publisherRecordRepository.findFirstByRoomIdAndUserIdAndPublisherStateInOrderByCreateTimeDesc(roomId, userId,
                Lists.newArrayList(PublisherStatus.APPLYING.getCode(), PUBLISHING.getCode()));

        if (liveRoomPublisherRecord != null && PUBLISHING.getCode() == liveRoomPublisherRecord.getPublisherState()) {
            redisTemplate.opsForValue().set(currentPublishingRecordKey, liveRoomPublisherRecord, 5, TimeUnit.MINUTES);
        }

        return liveRoomPublisherRecord;
    }

    /**
     * 最后一次上麦记录
     *
     * @param userId
     * @return
     */
    public LiveRoomPublisherRecord getLastPublishingRecord(Long userId) {
        LiveRoomPublisherRecord publisherRecord = publisherRecordRepository.findFirstByUserIdOrderByCreateTimeDesc(userId);
        if(Objects.isNull(publisherRecord)){
            LiveRoomPublisherRecordBackUpDO backup = liveRoomPublisherRecordBackupRepository.findFirstByUserIdOrderByCreateTimeDesc(userId);
            publisherRecord = LiveRoomPublisherRecordBackUpDO.from(backup);
        }
        return publisherRecord;
    }

    /**
     * 取消申请上麦
     *
     * @param roomId
     * @param userId
     * @return
     */
    public RoomPublisherVO cancelApplyPublish(Long roomId, Long userId) {

        UserVO userVO = getUserVO(userId);
        if (userVO == null) {
            traceLoger.info("cancelApplyPublish 房间【{}】用户【{}】取消申请上麦，用户不存在", roomId, userId);
            log.error("[cancelApplyPublish]用户不存在，roomId={},userId={}", roomId, userId);
            throw new CheckParamException("用户不存在!");
        }
        LiveRoomInfo roomInfo = roomService.getRoomInfoNoState(roomId);
        if (roomInfo == null) {
            traceLoger.info("cancelApplyPublish 房间【{}】用户【{}】取消申请上麦，房间不存在", roomId, userId);
            log.error("[agreeApplyPublish]房间不存在，roomId={},userId={}", roomId, userId);
            throw new CheckParamException("房间不存在或已关闭!");
        }
        LiveRoomPublisherRecord liveRoomPublisherRecord = getCurrentPublishingRecord(roomId, userId);
        if (liveRoomPublisherRecord == null || !PublisherStatus.isValidStatus(liveRoomPublisherRecord.getPublisherState())) {
            traceLoger.info("cancelApplyPublish 房间【{}】用户【{}】取消申请上麦，上麦记录已经结束或不存在", roomId, userId);
            log.error("[cancelApplyPublish]上麦申请不存在，roomId={},userId={}", roomId, userId);
            return null;
        }

        if (PUBLISHING.getCode() == liveRoomPublisherRecord.getPublisherState()) {
            traceLoger.info("cancelApplyPublish 房间【{}】用户【{}】取消申请上麦，上麦记录已经结束或不存在", roomId, userId);
            return null;
        }

        Date now = new Date();
        liveRoomPublisherRecord.setQuitTime(now);
        liveRoomPublisherRecord.setUpdateTime(now);
        liveRoomPublisherRecord.setPublisherState(PublisherStatus.CANCEL.getCode());

        liveRoomPublisherRecord = publisherRecordRepository.save(liveRoomPublisherRecord);
        traceLoger.info("cancelApplyPublish 房间【{}】用户【{}】取消申请上麦【{}】", roomId, userId, liveRoomPublisherRecord.getId());

        String currentPublishingRecordKey = RedisKeyConst.getCurrentPublishingRecordKey(userId, roomId);
        redisTemplate.delete(currentPublishingRecordKey);

        messageService.cancelApplyPublish(roomId, liveRoomPublisherRecord.getId(), userVO, liveRoomPublisherRecord.getApplyTime());

        RoomPublisherVO roomPublisherVO = new RoomPublisherVO();
        BeanUtils.copyProperties(liveRoomPublisherRecord, roomPublisherVO);
        return roomPublisherVO;

    }

    /**
     * 房主同意上麦
     *
     * @param roomId
     * @param userId
     * @param hostUserId
     * @return
     */
    public RoomPublisherVO agreeApplyPublish(Long roomId, Long userId, Long hostUserId) {


        UserVO userVO = getUserVO(userId);
        if (userVO == null) {
            traceLoger.info("agreeApplyPublish 房间【{}】房主【{}】同意用户【{}】上麦，用户不存在", roomId, hostUserId, userId);
            log.error("[agreeApplyPublish]用户不存在，roomId={},userId={}", roomId, userId);
            throw new CheckParamException("用户不存在!");
        }

        LiveRoomInfo roomInfo = roomService.getRoomInfoById(roomId);
        if (roomInfo == null) {
            traceLoger.info("agreeApplyPublish 房间【{}】房主【{}】同意用户【{}】上麦，房间不存在", roomId, hostUserId, userId);
            log.error("[agreeApplyPublish]房间不存在，roomId={},userId={}", roomId, userId);
            throw new CheckParamException("房间不存在或已关闭!");
        }
        if (!roomJoinRecordService.verifyOwner(roomId, hostUserId)) {
            traceLoger.info("agreeApplyPublish 房间【{}】房主【{}】同意用户【{}】上麦，房主当前已经不是房主", roomId, hostUserId, userId);
            throw new CheckParamException("您不是当前房间的房主!");
        }

        long publishingCount = getPublishingCount(roomId);
        if (publishingCount >= roomInfo.getRoomPublisherMax()) {
            traceLoger.info("agreeApplyPublish 房间【{}】房主【{}】同意用户【{}】上麦，当前房间主播已经满员", roomId, hostUserId, userId);
            throw new CheckParamException("当前房间主播已经满员!");
        }

        LiveRoomPublisherRecord liveRoomPublisherRecord = getCurrentPublishingRecord(roomId, userId);
        if (liveRoomPublisherRecord == null || !PublisherStatus.isValidStatus(liveRoomPublisherRecord.getPublisherState())) {
            log.error("[agreeApplyPublish]上麦申请不存在，roomId={},userId={}", roomId, userId);
            traceLoger.info("agreeApplyPublish 房间【{}】房主【{}】同意用户【{}】上麦，当前房间主播已经满员", roomId, hostUserId, userId);
            throw new BusinessException("上麦申请不存在!");
        }

        if (PublisherStatus.CANCEL.getCode() == liveRoomPublisherRecord.getPublisherState()) {
            traceLoger.info("agreeApplyPublish 房间【{}】房主【{}】同意用户【{}】上麦，当前申请记录【{}】已经取消", roomId, hostUserId, userId, liveRoomPublisherRecord.getId());
            throw new BusinessException("已经取消!");
        }
        if (PUBLISHING.getCode() == liveRoomPublisherRecord.getPublisherState()) {
            traceLoger.info("agreeApplyPublish 房间【{}】房主【{}】同意用户【{}】上麦，当前申请记录【{}】已经已经在上麦中", roomId, hostUserId, userId, liveRoomPublisherRecord.getId());
            throw new BusinessException("已经在上麦中!");
        }
        if (PublisherStatus.APPLYING.getCode() != liveRoomPublisherRecord.getPublisherState()) {
            traceLoger.info("agreeApplyPublish 房间【{}】房主【{}】同意用户【{}】上麦，当前申请记录【{}】状态【{}】不正确", roomId, hostUserId, userId,
                    liveRoomPublisherRecord.getId(), PublisherStatus.getPublisherStatus(liveRoomPublisherRecord.getPublisherState()).getTips());
            log.warn("[agreeApplyPublish]上麦申请状态不正确，liveRoomPublisherRecord={}", JSON.toJSONString(liveRoomPublisherRecord));
            throw new BusinessException("操作无效！");
        }

        Date now = new Date();
        liveRoomPublisherRecord.setAgreeTime(now);
        liveRoomPublisherRecord.setUpdateTime(now);
        liveRoomPublisherRecord.setPublisherState(PUBLISHING.getCode());
        liveRoomPublisherRecord = publisherRecordRepository.save(liveRoomPublisherRecord);

        traceLoger.info("agreeApplyPublish 房间【{}】房主【{}】同意用户【{}】上麦，当前申请记录【{}】", roomId, hostUserId, userId, liveRoomPublisherRecord.getId());
        roomJoinRecordService.roleChange(roomId, userId, RoomUserRole.PUBLISHER.getCode());
        roomJoinRecordService.micStateChange(roomId, userId, roomInfo.getRoomMicState());

        String publisherCountKey = RedisKeyConst.getRoomPublisherCountKey(roomId.toString());
        Long count = redisTemplate.opsForValue().increment(publisherCountKey, 1L);
        if (count > roomInfo.getRoomPublisherMax()) {
            redisTemplate.opsForValue().set(publisherCountKey, roomInfo.getRoomPublisherMax());
        }
        redisTemplate.expire(publisherCountKey, 2, TimeUnit.DAYS);
        redisTemplate.delete(Lists.newArrayList(RedisKeyConst.getRoomInfoByIdKey(roomId.toString()), RedisKeyConst.getCurrentPublishingRecordKey(userId, roomId)));
        clearLastPublish(roomId, userId);
        messageService.agreeApplyPublish(roomId, liveRoomPublisherRecord.getId(), hostUserId, userVO);

        liveTask(userId, roomId);

        LiveRoomPublisherRecord finalLiveRoomPublisherRecord = liveRoomPublisherRecord;
        asyncTaskExecutor.execute(() -> {
            startLearnTask(userId, roomId, finalLiveRoomPublisherRecord.getId(), finalLiveRoomPublisherRecord.getStudyContent());
        });

        RoomPublisherVO roomPublisherVO = new RoomPublisherVO();
        BeanUtils.copyProperties(liveRoomPublisherRecord, roomPublisherVO);
        return roomPublisherVO;

    }

    public void clearLastPublish(Long roomId, Long userId) {
        String userVideoUrlKey = RedisKeyConst.getUserVideoUrlKey(userId, roomId);
        String recordingPicDirKey = RedisKeyConst.getRecordingPicDirKey(userId, roomId);
        String studySecondsKey = RedisKeyConst.getStudySecondsKey(userId, roomId);
        String startLearnTaskKey = RedisKeyConst.getStartLearnTaskKey(userId, roomId);
        String stopLearnTaskKey = RedisKeyConst.getStopLearnTaskKey(userId, roomId);
        String lostLiveCountKey = RedisKeyConst.getLostLiveCountKey(userId, roomId);
        redisTemplate.delete(Lists.newArrayList(userVideoUrlKey, recordingPicDirKey, studySecondsKey, startLearnTaskKey, stopLearnTaskKey, lostLiveCountKey));
    }

    /**
     * 是否在连麦中
     *
     * @param roomId
     * @param userId
     * @return
     */
    public boolean isPublishing(Long roomId, Long userId) {
        LiveRoomPublisherRecord liveRoomPublisherRecord = getCurrentPublishingRecord(roomId, userId);
        return liveRoomPublisherRecord != null
                && PUBLISHING.getCode() == liveRoomPublisherRecord.getPublisherState();
    }

    public LiveRoomPublisherRecord save(LiveRoomPublisherRecord publishingRecord) {
        String currentPublishingRecordKey = RedisKeyConst.getCurrentPublishingRecordKey(publishingRecord.getUserId(), publishingRecord.getRoomId());
        redisTemplate.delete(currentPublishingRecordKey);
        return publisherRecordRepository.save(publishingRecord);
    }


    /**
     * 房主下麦之后查找下一个房主
     *
     * @param roomId
     * @return
     */
    public UserVO getNextHostUserId(Long roomId, Set<Long> notUids) {

        Set<Long> finalNotUids = notUids;
        Specification<LiveRoomPublisherRecord> querySpecifi = new Specification<LiveRoomPublisherRecord>() {

            @Override
            public Predicate toPredicate(Root<LiveRoomPublisherRecord> root, CriteriaQuery<?> criteriaQuery, CriteriaBuilder cb) {
                List<Predicate> predicates = new ArrayList<>();
                predicates.add(cb.equal(root.get("roomId").as(Long.class), roomId));
                predicates.add(cb.equal(root.get("publisherState").as(Integer.class), PUBLISHING.getCode()));
                predicates.add(cb.lessThan(root.get("applyTime").as(Date.class), DateUtils.addSeconds(new Date(), 30)));
                if (CollectionUtils.isNotEmpty(finalNotUids)) {
                    CriteriaBuilder.In<Long> notUidsIn = cb.in(root.get("userId"));
                    for (Long uid : finalNotUids) {
                        notUidsIn.value(uid);
                    }
                    predicates.add(cb.not(notUidsIn));
                }
                return cb.and(predicates.toArray(new Predicate[predicates.size()]));
            }
        };

        Sort sort = new Sort(Sort.Direction.ASC, "agreeTime");
        List<LiveRoomPublisherRecord> list = publisherRecordRepository.findAll(querySpecifi, sort);
        if (CollectionUtils.isNotEmpty(list)) {
//            String liveTaskKey = RedisKeyConst.getLiveTaskKey(list.get(0).getUserId(),list.get(0).getRoomId());
//            Date liveDate = (Date) redisTemplate.opsForValue().get(liveTaskKey);
//            if(liveDate!=null && new Date().before(DateUtils.addSeconds(liveDate,30))){
            for (LiveRoomPublisherRecord publisherRecord : list) {
                UserVO userVO = getUserVO(publisherRecord.getUserId());
                if (userVO != null) {
                    //检查是否是vip用户
                    LiveGatherVO gather = userClient.gather(publisherRecord.getUserId()).getData();
                    if (!gather.getIsCreateRoom()) {
                            continue;
                    }
                    return userVO;
                }
            }
//            }else {
//                if(notUids==null){
//                    notUids= Sets.newHashSet();
//                }
//                notUids.add(list.get(0).getUserId());
//                return getNextHostUserId( roomId,  notUids);
//            }
        }

        return null;
    }

    /**
     * 房主关闭房间下麦
     *
     * @param roomId
     */
    public void closeRoomQuitPublish(Long roomId) {

        traceLoger.info("closeRoomQuitPublish 房间【{}】关闭，上麦记录全部结束", roomId);
        List<LiveRoomPublisherRecord> publisherRecordList = getPublishingList(roomId);
        publisherRecordRepository.quitAll(roomId, new Date());

        if (CollectionUtils.isEmpty(publisherRecordList)) {
            return;
        }
        List<String> keys = new ArrayList<>();
        publisherRecordList.forEach(o -> {
            keys.add(RedisKeyConst.getCurrentPublishingRecordKey(o.getUserId(), roomId));
            stopLearnTask(o.getUserId(), roomId, o.getId(), o.getStudyContent());
        });
        keys.add(RedisKeyConst.getRoomPublisherCountKey(roomId.toString()));

        try {
            redisTemplate.delete(keys);
        } catch (Exception e) {
            log.error("closeRoomQuitPublish delete cache error ={}", e);
        }
    }

    public void finishPublish(Long roomId, Long userId, int closeState) {
        traceLoger.info("finishPublish 房间【{}】用户【{}】结束 上麦记录状态", roomId, userId);
        LiveRoomPublisherRecord liveRoomPublisherRecord = getCurrentPublishingRecord(roomId, userId);
        if (liveRoomPublisherRecord == null || !PublisherStatus.isValidStatus(liveRoomPublisherRecord.getPublisherState())) {
            return;
        }

        if (PublisherStatus.APPLYING.getCode() == liveRoomPublisherRecord.getPublisherState()) {
            cancelApplyPublish(roomId, userId);
        } else if (PUBLISHING.getCode() == liveRoomPublisherRecord.getPublisherState()) {
            quitPublish(roomId, userId, false, closeState);
        }
    }


    /**
     * 下麦
     *
     * @param roomId
     * @param userId
     * @return
     */
    public RoomPublisherVO quitPublish(Long roomId, Long userId, boolean onlyQuitPublish){
        return quitPublish(roomId, userId, onlyQuitPublish, RoomState.CLOSED.getCode());
    }

    /**
     * 下麦
     *
     * @param roomId
     * @param userId
     * @return
     */
    public RoomPublisherVO quitPublish(Long roomId, Long userId, boolean onlyQuitPublish, int closeState) {
        traceLoger.info("quitPublish 房间【{}】用户【{}】下麦,onlyQuit={}", roomId, userId, onlyQuitPublish);

        UserVO userVO = getUserVO(userId);
        if (userVO == null) {
            traceLoger.info("quitPublish 房间【{}】用户【{}】下麦,用户不存在", roomId, userId);
            log.error("[quitPublish]用户不存在，roomId={},userId={}", roomId, userId);
            throw new CheckParamException("用户不存在!");
        }

        LiveRoomPublisherRecord liveRoomPublisherRecord = getCurrentPublishingRecord(roomId, userId);
        if (liveRoomPublisherRecord == null || !PublisherStatus.isValidStatus(liveRoomPublisherRecord.getPublisherState())) {
            traceLoger.info("quitPublish 房间【{}】用户【{}】下麦,上麦记录不存在或已经结束", roomId, userId);
            log.error("[quitPublish]上麦记录不存在，roomId={},userId={}", roomId, userId);
            return null;
        }

        if (PUBLISHING.getCode() != liveRoomPublisherRecord.getPublisherState()) {
            traceLoger.info("quitPublish 房间【{}】用户【{}】下麦,上麦记录状态【{}】不正确", roomId, userId, PublisherStatus.getPublisherStatus(liveRoomPublisherRecord.getPublisherState()));
            log.warn("[quitPublish]上麦申请状态不正确，liveRoomPublisherRecord={}", JSON.toJSONString(liveRoomPublisherRecord));
            return null;
        }
        Date now = new Date();
        liveRoomPublisherRecord.setQuitTime(now);
        liveRoomPublisherRecord.setUpdateTime(now);
        liveRoomPublisherRecord.setPublisherState(PublisherStatus.QUIT.getCode());

        log.debug("liveRoomPublisherRecord = {}", liveRoomPublisherRecord);
        liveRoomPublisherRecord = publisherRecordRepository.save(liveRoomPublisherRecord);
        redisTemplate.delete(RedisKeyConst.getCurrentPublishingRecordKey(liveRoomPublisherRecord.getUserId(), roomId));

        traceLoger.info("quitPublish 房间【{}】用户【{}】已经下麦", roomId, userId);

        LiveRoomPublisherRecord finalLiveRoomPublisherRecord = liveRoomPublisherRecord;
//        asyncTaskExecutor.execute(()->{
        try {
            String publisherCountKey = RedisKeyConst.getRoomPublisherCountKey(roomId.toString());
            Long count = redisTemplate.opsForValue().increment(publisherCountKey, -1L);
            if (count <= 0) {
                redisTemplate.delete(publisherCountKey);
            } else {
                redisTemplate.expire(publisherCountKey, 2, TimeUnit.DAYS);
            }

            stopLearnTask(userId, roomId, finalLiveRoomPublisherRecord.getId(), finalLiveRoomPublisherRecord.getStudyContent());

            boolean isHost = roomJoinRecordService.verifyOwner(roomId, userId);
            if (isHost) {
                UserVO nextHostUserInfo = getNextHostUserId(roomId, Sets.newHashSet(userId));
                if (nextHostUserInfo == null) {
                    traceLoger.info("quitPublish 房间【{}】房主【{}】下麦之后无主播，房间关闭", roomId, userId);
                    //解散房间
                    roomService.closeRoom(roomId, closeState, userId);
                    log.info("[quitPublish]房主{}下麦之后无其他主播房间关闭", userId);
                } else {
                    traceLoger.info("quitPublish 房间【{}】房主【{}】下麦之后换【{}】为新房主", roomId, userId, nextHostUserInfo.getId());
                    //换房主
                    roomJoinRecordService.roleChange(roomId, userId, RoomUserRole.AUDIENCE.getCode());
                    roomJoinRecordService.roleChange(roomId, nextHostUserInfo.getId(), RoomUserRole.OWNER.getCode());

                    roomService.updateRoomAvatarAndDesc(roomId, nextHostUserInfo.getAvatarUrl());

                    messageService.hostChange(roomId, userId, nextHostUserInfo);
                    log.info("[quitPublish]房主{}下麦之后将{}切换成新房主", userId, nextHostUserInfo.getId());
                }
            } else {
                traceLoger.info("quitPublish 房间【{}】用户【{}】下麦成观众", roomId, userId);
                roomJoinRecordService.roleChange(roomId, userId, RoomUserRole.AUDIENCE.getCode());
                log.info("[quitPublish]用户{}下麦切换成观众", userId);
                if (onlyQuitPublish) {
                    messageService.quitPublish(roomId, userVO);
                }
            }

            redisTemplate.delete(Lists.newArrayList(RedisKeyConst.getRoomInfoByIdKey(roomId.toString()), RedisKeyConst.getCurrentPublishingRecordKey(finalLiveRoomPublisherRecord.getUserId(), roomId)));

        } catch (Exception e) {
            log.error("[quitPublish]部分操作失败，liveRoomPublisherRecord={},e={}", JSON.toJSONString(finalLiveRoomPublisherRecord), e);
        }
//        });

        RoomPublisherVO roomPublisherVO = new RoomPublisherVO();
        BeanUtils.copyProperties(liveRoomPublisherRecord, roomPublisherVO);
        return roomPublisherVO;

    }

    /**
     * 全员禁麦操作
     *
     * @param roomId
     * @param hostUserId
     * @param micState
     * @throws Exception
     */
    public void operateAllMic(Long roomId, Long hostUserId, Integer micState) throws Exception {

        if (RoomMicStatus.getRoomMicStatus(micState) == null) {
            traceLoger.info("operateAllMic 房间【{}】房主【{}】全员麦克操作状态【{}】不正确", roomId, hostUserId, micState);
            throw new CheckParamException("麦克操作状态不正确!");
        }

        UserVO userVO = getUserVO(hostUserId);
        if (userVO == null) {
            traceLoger.info("operateAllMic 房间【{}】房主【{}】全员麦克操作,房主不存在", roomId, hostUserId);
            log.error("[operateAllMic]用户不存在，roomId={},userId={}", roomId, hostUserId);
            throw new CheckParamException("用户不存在!");
        }

        LiveRoomInfo roomInfo = roomService.getRoomInfoById(roomId);
        if (roomInfo == null) {
            traceLoger.info("operateAllMic 房间【{}】房主【{}】全员麦克操作,房间不存在", roomId, hostUserId);
            log.error("[operateAllMic]房间不存在，roomId={},hostUserId={},micState={}", roomId, hostUserId, micState);
            throw new CheckParamException("房间不存在或已关闭!");
        }

        if (!roomJoinRecordService.verifyOwner(roomId, hostUserId)) {
            traceLoger.info("operateAllMic 房间【{}】房主【{}】全员麦克操作,不是房主", roomId, hostUserId);
            throw new CheckParamException("您不是当前房间的房主!");
        }
        traceLoger.info("operateAllMic 房间【{}】房主【{}】全员麦克操作状态【{}】", roomId, hostUserId, RoomMicStatus.getRoomMicStatus(micState).getTips());
        roomService.updateRoomMicState(roomId, micState, hostUserId);
        messageService.operateAllMic(roomId, micState, userVO);

    }

    /**
     * 自己麦克操作
     *
     * @param roomId
     * @param userId
     * @param micState
     * @throws Exception
     */
    public void operateSelfMic(Long roomId, Long userId, Integer micState) {

        if (RoomMicStatus.getRoomMicStatus(micState) == null) {
            traceLoger.info("operateSelfMic 房间【{}】用户【{}】麦克操作状态【{}】不正确", roomId, userId, micState);
            throw new CheckParamException("麦克操作状态不正确!");
        }
        UserVO userVO = getUserVO(userId);
        if (userVO == null) {
            traceLoger.info("operateSelfMic 房间【{}】用户【{}】麦克操作状态【{}】，用户不存在", roomId, userId, micState);
            log.error("[operateSelfMic]用户不存在，roomId={},userId={}", roomId, userId);
            throw new CheckParamException("用户不存在!");
        }
        LiveRoomInfo roomInfo = roomService.getRoomInfoById(roomId);
        if (roomInfo == null) {
            traceLoger.info("operateSelfMic 房间【{}】用户【{}】麦克操作状态【{}】，房间不存在", roomId, userId, micState);
            log.error("[operateSelfMic]房间不存在，roomId={},userId={},micState={}", roomId, userId, micState);
            throw new CheckParamException("房间不存在或已关闭!");
        }

        LiveRoomPublisherRecord liveRoomPublisherRecord = getCurrentPublishingRecord(roomId, userId);
        if (liveRoomPublisherRecord == null || !PublisherStatus.isValidStatus(liveRoomPublisherRecord.getPublisherState())) {
            traceLoger.info("operateSelfMic 房间【{}】用户【{}】麦克操作状态【{}】，上麦记录不存在或者已经结束", roomId, userId, micState);
            log.error("[operateSelfMic]上麦记录不存在，roomId={},userId={}", roomId, userId);
            throw new CheckParamException("当前未在连麦中!");
        }
        if ((roomInfo.getRoomMicState() == RoomMicStatus.MIC_OFF.getCode()
                && micState == RoomMicStatus.MIC_ON.getCode()) && !roomJoinRecordService.verifyOwner(roomId, userId)) {
            traceLoger.info("operateSelfMic 房间【{}】用户【{}】麦克操作状态【{}】，当前房间在禁麦中", roomId, userId, micState);
            throw new CheckParamException("当前房间在禁麦中!");
        }
        traceLoger.info("operateSelfMic 房间【{}】用户【{}】麦克操作状态【{}】", roomId, userId, RoomMicStatus.getRoomMicStatus(micState).getTips());
        roomJoinRecordService.micStateChange(roomId, userId, micState);
        messageService.operateSelfMic(roomId, micState, userVO);
    }


    /**
     * 上麦申请列表
     *
     * @param roomId
     * @param hostUserId
     */
    public List<RoomPublisherVO> applyList(Long roomId, Long hostUserId) {

        if (!roomJoinRecordService.verifyOwner(roomId, hostUserId)) {
            throw new CheckParamException("您不是当前房间的房主!");
        }


        String cancelPublishingRecordSetKey = RedisKeyConst.getCancelPublishingRecordSetKey(roomId);
        Set<Integer> canceledIdSet = redisTemplate.opsForSet().members(cancelPublishingRecordSetKey);

        Specification<LiveRoomPublisherRecord> querySpecifi = new Specification<LiveRoomPublisherRecord>() {

            @Override
            public Predicate toPredicate(Root<LiveRoomPublisherRecord> root, CriteriaQuery<?> criteriaQuery, CriteriaBuilder cb) {
                List<Predicate> predicates = new ArrayList<>();
                predicates.add(cb.equal(root.get("roomId").as(Long.class), roomId));
                predicates.add(cb.in(root.get("publisherState")).value(PublisherStatus.APPLYING.getCode()).value(PublisherStatus.CANCEL.getCode()));
//                   predicates.add(cb.lessThan(root.get("applyTime").as(Date.class), DateUtils.addSeconds(new Date(),30)));

                if (CollectionUtils.isNotEmpty(canceledIdSet)) {
                    CriteriaBuilder.In<Long> notInIds = cb.in(root.get("id"));
                    for (Integer id : canceledIdSet) {
                        notInIds.value(Long.valueOf(id));
                    }
                    predicates.add(cb.not(notInIds));
                }

                return cb.and(predicates.toArray(new Predicate[predicates.size()]));
            }
        };

        Sort sort = new Sort(Lists.newArrayList(new Sort.Order(Sort.Direction.ASC, "publisherState"), new Sort.Order(Sort.Direction.DESC, "applyTime")));
        List<LiveRoomPublisherRecord> list = publisherRecordRepository.findAll(querySpecifi, sort);
        List<RoomPublisherVO> publisherVOList = Lists.newArrayList();
        if (CollectionUtils.isEmpty(list)) {
            return publisherVOList;
        }
//          boolean addCancel = false;
        for (LiveRoomPublisherRecord roomPublisherRecord : list) {
//              if(PublisherStatus.CANCEL.getCode()==roomPublisherRecord.getPublisherState()){
//                  redisTemplate.opsForSet().add(cancelPublishingRecordSetKey,roomPublisherRecord.getId());
//                  addCancel = true;
//              }
            RoomPublisherVO publisherVO = new RoomPublisherVO();
            BeanUtils.copyProperties(roomPublisherRecord, publisherVO);
            UserVO userVO = getUserVO(publisherVO.getUserId());
            if (userVO != null) {
                publisherVO.setUserName(userVO.getName());
                publisherVO.setUserAvatarUrl(userVO.getAvatarUrl());
            }
            publisherVOList.add(publisherVO);
        }
//          if(addCancel){
//              redisTemplate.expire(cancelPublishingRecordSetKey,2,TimeUnit.DAYS);
//          }

        return publisherVOList;
    }

    public void ackCancelApplyPublish(Long roomId, Long hostUserId, List<Long> ackCancelApplyIdList) {

        if (!roomJoinRecordService.verifyOwner(roomId, hostUserId)) {
            throw new CheckParamException("您不是当前房间的房主!");
        }

        if (CollectionUtils.isNotEmpty(ackCancelApplyIdList)) {
            Specification<LiveRoomPublisherRecord> querySpecifi = new Specification<LiveRoomPublisherRecord>() {
                @Override
                public Predicate toPredicate(Root<LiveRoomPublisherRecord> root, CriteriaQuery<?> criteriaQuery, CriteriaBuilder cb) {
                    List<Predicate> predicates = new ArrayList<>();
                    predicates.add(cb.equal(root.get("roomId").as(Long.class), roomId));
                    predicates.add(cb.equal(root.get("publisherState").as(Integer.class), PublisherStatus.CANCEL.getCode()));
                    CriteriaBuilder.In<Long> idsIn = cb.in(root.get("id"));
                    for (Long id : ackCancelApplyIdList) {
                        idsIn.value(id);
                    }
                    predicates.add(idsIn);

                    return cb.and(predicates.toArray(new Predicate[predicates.size()]));
                }
            };
            List<LiveRoomPublisherRecord> list = publisherRecordRepository.findAll(querySpecifi);
            List<Long> ids = list.stream().map(LiveRoomPublisherRecord::getId).collect(Collectors.toList());
            if (CollectionUtils.isNotEmpty(ids)) {
                String cancelPublishingRecordSetKey = RedisKeyConst.getCancelPublishingRecordSetKey(roomId);
                redisTemplate.opsForSet().add(cancelPublishingRecordSetKey, ids.toArray(new Long[]{}));
                redisTemplate.expire(cancelPublishingRecordSetKey, 2, TimeUnit.DAYS);
            }
        }

    }


    /**
     * 邀请列表
     *
     * @param userId
     * @return
     */
    public List<InviteListVO> inviteList(Long userId) {
        if (Objects.isNull(userId) || userId == 0) {
            throw new CheckParamException("用户id不正确");
        }

        List<InviteListVO> list = Lists.newArrayList();
        List<GroupVO> groupVOList = groupService.myGroupList(userId);
        if (CollectionUtils.isNotEmpty(groupVOList)) {
            for (GroupVO groupVO : groupVOList) {
                InviteListVO inviteListVO = new InviteListVO();
                inviteListVO.setGroupId(groupVO.getGroupId());
                inviteListVO.setName(groupVO.getGroupName());
                inviteListVO.setAvatarUrl(groupVO.getAvatarFile());
                list.add(inviteListVO);
            }
        }
        List<UserFriendVO> friendVOList = userService.userFriendsList(userId);
        if (CollectionUtils.isNotEmpty(friendVOList)) {
            for (UserFriendVO friendVO : friendVOList) {
                if (userId.equals(friendVO.getFriendsUserId())) {
                    continue;
                }
                InviteListVO inviteListVO = new InviteListVO();
                inviteListVO.setUserId(friendVO.getFriendsUserId());
                inviteListVO.setName(friendVO.getFriendsUserNickName());
                inviteListVO.setAvatarUrl(friendVO.getFriendsUserAvatar());
                list.add(inviteListVO);
            }
        }

        return list;
    }

    /**
     * 邀请道友和群友上麦
     *
     * @param roomId
     * @param userId
     * @param shareToUserIds
     * @param shareToGroupIds
     * @throws Exception
     */
    public void invitePublish(Long roomId, Long userId, List<Long> shareToUserIds, List<String> shareToGroupIds) {

        LiveRoomInfo roomInfo = roomService.getRoomInfoById(roomId);
        if (roomInfo == null) {
            log.error("[invitePublish]房间不存在，roomId={},userId={}", roomId, userId);
            throw new CheckParamException("房间不存在或已关闭!");
        }
        if (roomJoinRecordService.getUserJoinByRoomIdAndUserId(roomId, userId) == null) {
            throw new CheckParamException("未进入此房间!");
        }

        messageService.invitePublish(roomId, userId, shareToUserIds, shareToGroupIds, roomInfo);

    }

    /**
     * 当前上麦人数
     *
     * @param roomId
     * @return
     */
    public long getPublishingCount(Long roomId) {
        String publisherCountKey = RedisKeyConst.getRoomPublisherCountKey(roomId.toString());
        Object n = redisTemplate.opsForValue().get(publisherCountKey);
        if (n != null) {
            return Long.parseLong(n.toString());
        }

        LiveRoomPublisherRecord recordparams = new LiveRoomPublisherRecord();
        recordparams.setRoomId(roomId);
        recordparams.setPublisherState(PUBLISHING.getCode());
        Example<LiveRoomPublisherRecord> example = Example.of(recordparams);

        long count = publisherRecordRepository.count(example);
        redisTemplate.opsForValue().set(publisherCountKey, count, 1, TimeUnit.HOURS);
        return count;
    }

    public List<LiveRoomPublisherRecord> getPublishingList(Long roomId) {

        LiveRoomPublisherRecord recordparams = new LiveRoomPublisherRecord();
        recordparams.setRoomId(roomId);
        recordparams.setPublisherState(PUBLISHING.getCode());
        Example<LiveRoomPublisherRecord> example = Example.of(recordparams);

        return publisherRecordRepository.findAll(example);
    }

    /**
     * 查询上麦时间
     *
     * @param userId
     * @param roomId
     * @return
     */
    public Date getStartPublishingDate(Long userId, Long roomId) {

        String key = RedisKeyConst.getStartLearnTaskKey(userId, roomId);
        Date startDate = (Date) redisTemplate.opsForValue().get(key);

        if (startDate != null) {
            return startDate;
        }
        LiveRoomPublisherRecord liveRoomPublisherRecord = getCurrentPublishingRecord(roomId, userId);
        if (liveRoomPublisherRecord != null) {
            return liveRoomPublisherRecord.getAgreeTime();
        }

        return null;
    }

    public Date getStopPublishingDate(Long userId, Long roomId) {

        String key = RedisKeyConst.getStopLearnTaskKey(userId, roomId);
        Date stopDate = (Date) redisTemplate.opsForValue().get(key);

        if (stopDate != null) {
            return stopDate;
        }
        LiveRoomPublisherRecord liveRoomPublisherRecord = getCurrentPublishingRecord(roomId, userId);
        if (liveRoomPublisherRecord != null) {
            return liveRoomPublisherRecord.getQuitTime();
        }

        return null;
    }


    public void startLearnTask(Long userId, Long roomId, Long publishRecordId, String studyContent) {
        traceLoger.info("startLearnTask 房间【{}】用户【{}】上麦【{}】后 开始学习", roomId, userId, publishRecordId);
        Assert.notNull(userId, "userId不能为空");
        Assert.notNull(roomId, "roomId不能为空");
        Assert.notNull(publishRecordId, "publishRecordId不能为空");

        LiveRoomInfo roomInfo = roomService.getRoomInfoNoState(roomId);
        if (roomInfo == null) {
            traceLoger.info("startLearnTask 房间【{}】用户【{}】上麦【{}】后 开始学习，房间不存在", roomId, userId, publishRecordId);
            log.error("[startLearnTask]房间不存在，roomId={},userId={}", roomId, userId);
            return;
        }

        Date now = new Date();
        String key = RedisKeyConst.getStartLearnTaskKey(userId, roomId);
        redisTemplate.opsForValue().set(key, now, 1, TimeUnit.HOURS);
        redisTemplate.delete(RedisKeyConst.getStopLearnTaskKey(userId, roomId));
        LearnTimeParamVO learnTimeParamVO = new LearnTimeParamVO();
        learnTimeParamVO.setUserId(userId);
        learnTimeParamVO.setType(15);
        learnTimeParamVO.setStudyInfoName(StringUtils.isBlank(studyContent) ? roomInfo.getRoomName() : studyContent);
        learnTimeParamVO.setLianMaiId(publishRecordId);
        learnTimeParamVO.setRoomId(roomId);
        learnTimeParamVO.setStartTime(DateFormatUtils.format(now, "yyyy-MM-dd HH:mm:ss"));

        userService.addUserLearnTime(learnTimeParamVO, true);
    }

    public void stopLearnTask(Long userId, Long roomId, Long publishRecordId, String studyContent) {
        traceLoger.info("stopLearnTask 房间【{}】用户【{}】下麦【{}】，停止学习", roomId, userId, publishRecordId);
        log.info("[stopLearnTask] roomId={},userId={},publishRecordId={} studyContent={}", roomId, userId, publishRecordId, studyContent);
        Assert.notNull(userId, "userId不能为空");
        Assert.notNull(roomId, "roomId不能为空");
        Assert.notNull(publishRecordId, "publishRecordId不能为空");
        LiveRoomInfo roomInfo = roomService.getRoomInfoNoState(roomId);
        if (roomInfo == null) {
            traceLoger.info("stopLearnTask 房间【{}】用户【{}】下麦【{}】，房间不存在", roomId, userId, publishRecordId);
            log.error("[stopLearnTask]房间不存在，roomId={},userId={}", roomId, userId);
            return;
        }

        Date startDate = getStartPublishingDate(userId, roomId);
        if (startDate == null) {
            startDate = new Date();
        }

        Date now = new Date();
        Long studySeconds = (now.getTime() - startDate.getTime()) / 1000;

//        String stopLearnTaskKey = RedisKeyConst.getStopLearnTaskKey(userId,roomId);
//        redisTemplate.opsForValue().set(stopLearnTaskKey,now,2,TimeUnit.DAYS);
//        String studySecondsKey = RedisKeyConst.getStudySecondsKey(userId,roomId);
//        redisTemplate.opsForValue().set(studySecondsKey,studySeconds,2,TimeUnit.DAYS);

        LearnTimeParamVO learnTimeParamVO = new LearnTimeParamVO();
        learnTimeParamVO.setEndTime(DateFormatUtils.format(now, "yyyy-MM-dd HH:mm:ss"));
        learnTimeParamVO.setStartTime(DateFormatUtils.format(startDate, "yyyy-MM-dd HH:mm:ss"));
        learnTimeParamVO.setStudyInfoName(StringUtils.isBlank(studyContent) ? roomInfo.getRoomName() : studyContent);
        learnTimeParamVO.setStudyMinuteSeconds(studySeconds);
        learnTimeParamVO.setUserId(userId);
        learnTimeParamVO.setType(5);
        learnTimeParamVO.setLianMaiId(publishRecordId);
        learnTimeParamVO.setRoomId(roomId);
        log.info("记录学习记录 learnTimeParamVO = {}", learnTimeParamVO);
        userService.addUserLearnTime(learnTimeParamVO, false);

    }


    /**
     * 强制结束用户学习状态
     *
     * @param publishRecord publishRecord
     */
    public void forceStopLearnTask(LiveRoomPublisherRecord publishRecord) {
        traceLoger.info("forceStopLearnTask 强制结束用户学习状态【{}】，", JSON.toJSONString(publishRecord));
        log.info("强制结束用户连麦状态 publishRecord = {}", publishRecord);
        Long userId = publishRecord.getUserId();
        Long roomId = publishRecord.getRoomId();
        Assert.notNull(userId, "userId不能为空");
        Assert.notNull(roomId, "roomId不能为空");
        LiveRoomInfo roomInfo = roomService.getRoomInfoNoState(roomId);
        if (roomInfo == null) {
            log.error("[stopLearnTask]房间不存在，roomId={},userId={}", roomId, userId);
            return;
        }

        Date startDate = getStartPublishingDate(userId, roomId);
        if (startDate == null) {
            startDate = new Date();
        }

        Date now = new Date();
        Long studySeconds = (now.getTime() - startDate.getTime()) / 1000;

//        String stopLearnTaskKey = RedisKeyConst.getStopLearnTaskKey(userId,roomId);
//        redisTemplate.opsForValue().set(stopLearnTaskKey,now,2,TimeUnit.DAYS);
//        String studySecondsKey = RedisKeyConst.getStudySecondsKey(userId,roomId);
//        redisTemplate.opsForValue().set(studySecondsKey,studySeconds,2,TimeUnit.DAYS);

        LearnTimeParamVO learnTimeParamVO = new LearnTimeParamVO();
        learnTimeParamVO.setEndTime(DateFormatUtils.format(now, "yyyy-MM-dd HH:mm:ss"));
        learnTimeParamVO.setStartTime(DateFormatUtils.format(startDate, "yyyy-MM-dd HH:mm:ss"));
        learnTimeParamVO.setStudyInfoName(roomInfo.getRoomName());
        learnTimeParamVO.setStudyMinuteSeconds(studySeconds);
        learnTimeParamVO.setUserId(userId);
        learnTimeParamVO.setType(5);
        learnTimeParamVO.setLianMaiId(publishRecord.getId());
        learnTimeParamVO.setRoomId(roomId);
        log.info("记录学习记录 learnTimeParamVO = {}", learnTimeParamVO);
        userService.addUserLearnTime(learnTimeParamVO, false);

    }

    /**
     * //todo 将来用的话需要优化
     * 查询当前学习时长
     *
     * @param userId
     * @param roomId
     * @return
     */
    public Long studySeconds(Long userId, Long roomId) {
        String studySecondsKey = RedisKeyConst.getStudySecondsKey(userId, roomId);
        Integer studySeconds = (Integer) redisTemplate.opsForValue().get(studySecondsKey);
        return studySeconds == null ? 0L : studySeconds;
    }

    public JSONObject liveTask(Long userId, Long roomId) {

        JSONObject jsonObject = new JSONObject();
        if (!roomJoinRecordService.isUserJoin(roomId, userId)) {
            if (Objects.nonNull(redisTemplate.opsForValue().get(RedisKeyConst.getRoomCloseDelayStatusKey(roomId)))) {
                traceLoger.info("liveTask 房间【{}】已经关闭，顺延用户【{}】心跳状态【{}】，", roomId, userId, YN.YES.getTips());
                jsonObject.put("inRoomState", YN.YES.getCode());
            } else {
                traceLoger.info("liveTask 房间【{}】用户【{}】心跳状态【{}】，", roomId, userId, YN.NO.getTips());
                jsonObject.put("inRoomState", YN.NO.getCode());
            }
            return jsonObject;
        }
        String key = RedisKeyConst.getLiveTaskKey(userId, roomId);
        redisTemplate.opsForValue().set(key, new Date(), liveTimeOut, TimeUnit.SECONDS);
        jsonObject.put("inRoomState", YN.YES.getCode());
        redisTemplate.delete(RedisKeyConst.getLostLiveCountKey(userId, roomId));
        traceLoger.info("liveTask 房间【{}】用户【{}】心跳状态【{}】，", roomId, userId, YN.YES.getTips());
        return jsonObject;
    }

    public StatusVO statusInfo(Long userId, Long roomId) {
        StatusVO statusVO = new StatusVO();
        if (roomId != null) {
            statusVO.setRoomValidState(YN.NO.getCode());
            LiveRoomInfo roomInfo = roomService.getRoomInfoById(roomId);
            if (roomInfo != null) {
                statusVO.setRoomValidState(YN.YES.getCode());
                statusVO.setRoomName(roomInfo.getRoomName());
            }
        }

        if (roomId != null && userId != null) {
            UserJoinVO userJoinVO = roomJoinRecordService.getUserJoinByRoomIdAndUserId(roomId, userId);
            if (userJoinVO == null) {
                statusVO.setInRoomState(YN.NO.getCode());
                statusVO.setUserRole(-1);
                statusVO.setPublisherState(-1);
            } else {
                statusVO.setInRoomState(YN.YES.getCode());
                statusVO.setUserRole(userJoinVO.getUserRole());
                statusVO.setPublisherState(-1);
                statusVO.setRoomId(userJoinVO.getRoomId());
                if (RoomUserRole.OWNER.getCode() == userJoinVO.getUserRole() || RoomUserRole.PUBLISHER.getCode() == userJoinVO.getUserRole()) {
                    LiveRoomPublisherRecord liveRoomPublisherRecord = getCurrentPublishingRecord(roomId, userId);
                    if (liveRoomPublisherRecord != null) {
                        statusVO.setPublisherState(liveRoomPublisherRecord.getPublisherState());
                    }
                }
                if (!roomId.equals(userJoinVO.getRoomId())) {
                    statusVO.setRoomValidState(YN.NO.getCode());
                    LiveRoomInfo roomInfo = roomService.getRoomInfoById(userJoinVO.getRoomId());
                    if (roomInfo != null) {
                        statusVO.setRoomValidState(YN.YES.getCode());
                        statusVO.setRoomName(roomInfo.getRoomName());
                    }
                }
            }
        } else if (roomId == null && userId != null) {
            LiveRoomJoinRecord roomJoinRecord = roomJoinRecordService.getUserJoinRecord(userId);
            if (roomJoinRecord == null) {
                statusVO.setInRoomState(YN.NO.getCode());
                statusVO.setUserRole(-1);
                statusVO.setPublisherState(-1);
            } else {
                statusVO.setInRoomState(YN.YES.getCode());
                statusVO.setUserRole(roomJoinRecord.getUserRole());
                statusVO.setPublisherState(-1);
                statusVO.setRoomId(roomJoinRecord.getRoomId());
                if (RoomUserRole.OWNER.getCode() == roomJoinRecord.getUserRole() || RoomUserRole.PUBLISHER.getCode() == roomJoinRecord.getUserRole()) {
                    LiveRoomPublisherRecord liveRoomPublisherRecord = getCurrentPublishingRecord(roomJoinRecord.getRoomId(), userId);
                    if (liveRoomPublisherRecord != null) {
                        statusVO.setPublisherState(liveRoomPublisherRecord.getPublisherState());
                    }
                }
                statusVO.setRoomValidState(YN.NO.getCode());
                LiveRoomInfo roomInfo = roomService.getRoomInfoById(roomJoinRecord.getRoomId());
                if (roomInfo != null) {
                    statusVO.setRoomValidState(YN.YES.getCode());
                    statusVO.setRoomName(roomInfo.getRoomName());
                }
            }
        }
        traceLoger.info("statusInfo 房间【{}】用户【{}】状态详情【{}】，", roomId, userId, JSON.toJSONString(statusVO));
        return statusVO;
    }


    public List<LiveRoomPublisherRecord> getNotNonalQuitPublisherList() {
        return publisherRecordRepository.getNotNonalQuitPublisher();
    }

    public List<LiveRoomPublisherRecord> getNoCancelTimeoutRecord() {
        return publisherRecordRepository.getNoCancelTimeoutRecord(DateUtils.addMinutes(new Date(), -1));
    }


    private UserVO getUserVO(Long userId) {
        return userService.getUserInfo(userId);
    }


    public RoomPublisherVO findPublisherRecord(Long publishRecordId) {
        LiveRoomPublisherRecord publisherRecord = publisherRecordRepository.findOne(publishRecordId);
        if (publisherRecord == null) {
            log.error("[findPublisherRecord]上麦记录不存在，publishRecordId={}", publishRecordId);
            throw new CheckParamException("上麦记录不存在");
        }

        RoomPublisherVO roomPublisherVO = new RoomPublisherVO();
        BeanUtils.copyProperties(publisherRecord, roomPublisherVO);
        return roomPublisherVO;
    }


    public List<LiveRoomPublisherRecord> findByRoomId(Long id) {
        return publisherRecordRepository.findByRoomId(id);
    }

    public void delete(Long id) {
        publisherRecordRepository.delete(id);
    }

    public RoomPublisherVO findPublisherRecordByUserId(Long userId){
        LiveRoomPublisherRecord publisherState = publisherRecordRepository.findFirstByUserIdAndPublisherState(userId, PUBLISHING.getCode());
        if(publisherState == null){
            return null;
        }
        RoomPublisherVO roomPublisherVO = new RoomPublisherVO();
        BeanUtils.copyProperties(publisherState, roomPublisherVO);
        return roomPublisherVO;
    }

    public List<RoomPublisherVO> findRoomListByUserIds(List<Long> userIds){
        List<LiveRoomPublisherRecord> publisher = publisherRecordRepository.findAllByUserIdInAndPublisherState(userIds, PUBLISHING.getCode());
        List<RoomPublisherVO> roomPublisherVOS = new ArrayList<>();
        publisher.forEach(liveRoomPublisherRecord -> {
            RoomPublisherVO roomPublisherVO = new RoomPublisherVO();
            BeanUtils.copyProperties(liveRoomPublisherRecord, roomPublisherVO);
            roomPublisherVOS.add(roomPublisherVO);
        });
        return roomPublisherVOS;
    }
}
